package com.qf.listener;

import com.fasterxml.jackson.databind.ObjectMapper;
import com.qf.constant.RabbitMQConstants;
import com.qf.model.StandardSubmit;
import com.rabbitmq.client.Channel;
import lombok.extern.slf4j.Slf4j;
import org.elasticsearch.action.index.IndexRequest;
import org.elasticsearch.action.index.IndexResponse;
import org.elasticsearch.client.RequestOptions;
import org.elasticsearch.client.RestHighLevelClient;
import org.elasticsearch.common.xcontent.XContentType;
import org.springframework.amqp.core.Message;
import org.springframework.amqp.core.Queue;
import org.springframework.amqp.rabbit.annotation.RabbitListener;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.stereotype.Component;
import reactor.util.concurrent.Queues;

import java.io.IOException;

/**
 * @author WzYu
 * @date 2022-6-10 下午 10:22
 * 监听rabbitMQ中 sms_send_Log
 */
@Component   //@component：  标注一个类为Spring容器的Bean，（把普通pojo实例化到spring容器中，相当于配置文件中的<bean id="" class=""/>）
@Slf4j
public class LogListener {


    /**
     * 启动器 自动装配
     */
    @Autowired
    private RestHighLevelClient restHighLevelClient;// ElasticSearch的客户端（高级别）
    @Value("${log-index}")
    private String indexName;
    @Autowired
    private ObjectMapper objectMapper;

    /**
     * 监听开始
     */
    @RabbitListener(queues = RabbitMQConstants.TOPIC_SMS_SEND_LOG)
    public void receiverSmsSendLog(Message message, Channel channel,StandardSubmit submit){

        try {
            log.info("ES微服务接收sms_send_log的日至：");

            IndexRequest indexRequest = new IndexRequest(indexName);//1:ES 保存文档
            indexRequest.id(submit.getSequenceId());//主键id
            String submitJson = objectMapper.writeValueAsString(submit);
            indexRequest.source(submitJson, XContentType.JSON);//文档
            IndexResponse indexResponse = restHighLevelClient.index(indexRequest, RequestOptions.DEFAULT);//执行保存
            System.out.println("结果:"+indexResponse.getResult().toString());
            if(indexResponse.getResult().toString().equalsIgnoreCase("created")
                    || indexResponse.getResult().toString().equalsIgnoreCase("updated") ){
                //成功  添加日志失败！！ result = UPDATED,su
                log.error("【搜索模块】添加日志 ， 添加日志成功！！ result = {},submit = {}",
                        indexResponse.getResult().toString(),submitJson);
                channel.basicAck(message.getMessageProperties().getDeliveryTag(),false);
            }else{
                log.error("【搜索模块】添加日志 ， 添加日志失败！！ result = {},submit = {}",
                        indexResponse.getResult().toString(),submitJson);
                //要求重新发送
                channel.basicNack(message.getMessageProperties().getDeliveryTag(),false,true);
            }
            channel.basicAck(message.getMessageProperties().getDeliveryTag(),false);
        } catch (Exception e) {
            e.printStackTrace();
        }


    }
}
